package com.study.listenMessage;

import com.study.Entity.Command;
import com.study.Entity.MqMessage;
import com.study.config.RocketMqConsumerConfiguration;
import com.study.send.SendFactory;
import com.study.send.Sender;
import com.study.utils.ByteObjectUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;

import java.util.List;

/**
 * @author mjy
 * @date 2021/12/16
 * @description
 */
public class RocketDealMessage extends RocketMqConsumerConfiguration implements ApplicationRunner {

    Logger log = LoggerFactory.getLogger(RocketMqConsumerConfiguration.class);

    /**
     * 消息处理
     *
     * @param msgs
     * @return
     */
    @Override
    public ConsumeConcurrentlyStatus dealMessage(List<MessageExt> msgs) {
        for (MessageExt msg : msgs) {
            MqMessage mqMessage = transformMsg(msg);
            log.info("接收到队列消息：" + mqMessage.toString());
            Sender sender = SendFactory.createSender(mqMessage);
            Command command = sender.dealMessage(mqMessage);
            boolean send = sender.send(command);
            log.info("消息队列发送："+(send?"成功":"失败"));
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }


    /**
     * 消息对象转化
     *
     * @param msg
     * @return
     */
    private MqMessage transformMsg(MessageExt msg) {
        MqMessage mqMessage = (MqMessage) ByteObjectUtils.toObject(msg.getBody());
        return mqMessage;
    }


    @Override
    public void run(ApplicationArguments args) throws Exception {
        try {
            log.info("程序启动后开始创建Mq监听");
            super.createConsumer();
            log.info("程序监听创建完成");
        } catch (MQClientException e) {
            log.info("程序监听创建异常");
            e.printStackTrace();
        }
    }
}
